package com.hadoop.seq;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.MapFile.Reader;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapred.MapFileOutputFormat;
import org.apache.hadoop.mapred.Partitioner;
import org.apache.hadoop.mapred.lib.HashPartitioner;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

//맵파일에서 원하는 데이터만 검색하면 되기 때문에 맵리듀스로 구현할 필요 없음
//검색하고자 하는 키가 속하는 파티션 번호를 조회한 후, 파티션 번호로 맵파일에 접근해 데이터를 검색.
public class SearchValueList extends Configured implements Tool {

	public int run(String[] args) throws Exception {
		Path path = new Path(args[0]);
		FileSystem fs = path.getFileSystem(getConf());

		// MapFile 조회 
		Reader[] readers = MapFileOutputFormat.getReaders(fs, path, getConf());

		// 검색 키를 저장할 객체를 선언
		IntWritable key = new IntWritable();
		key.set(Integer.parseInt(args[1]));

		// 검색 값을 저장할 객체를 선언 
		Text value = new Text();

		// 파티셔너를 이용해 검색 키가 저장된 MapFile 조회 
		// 만들어지 맵파일이 해시 파티셔너로 파티셔닝 됐기 때문에 해시 파티셔너로 생성
		Partitioner<IntWritable, Text> partitioner = new HashPartitioner<IntWritable, Text>();
		Reader reader = readers[partitioner.getPartition(key, value,
				readers.length)];

		// 검색 결과 확인 
		Writable entry = reader.get(key, value);
		if (entry == null) {
			System.out.println("The requested key was not found.");
		}

		//MapFile을 순회하며 키와 값을 출력
		IntWritable nextKey = new IntWritable();
		do {
			System.out.println(value.toString());
		} while (reader.next(nextKey, value) && key.equals(nextKey));

		return 0;
	}

	//조회하고자하는 해당폴더에 로그 파일을 읽다가 Exception 이 날 수 있으므로 로그 파일을 삭제 후 조회 (_SUCCESS, _logs 등)
	
	//상위 10개만 조회 
	// hadoop jar packagename.jar packagename.MainClass mapfile_folder명 <찾고자하는 Key값> | head -10
	public static void main(String[] args) throws Exception {
		int res = ToolRunner.run(new Configuration(), new SearchValueList(),
				args);
		System.out.println("## RESULT:" + res);
	}
}
